package com.zefor.utils;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zefor.pojo.ElectricityPojo;
import com.zefor.pojo.EnvironmentPojo;
import com.zefor.pojo.WaterPojo;
import org.apache.pulsar.client.api.*;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @version V1.0
 * @作者: 16847
 * @创建时间: 2023/5/22 11:00
 * @描述:
 * @包位置: com.zefor.utils
 */
/*
 *
 * Admq操作工具
 * */
public class AdmqUtil {

    // admq客户端
    private static PulsarClient pulsarClient = null;

    // jackson转换工具
    private static final ObjectMapper mapper = new ObjectMapper();


    static {
        // 配置admq地址信息
        try {
            pulsarClient = PulsarClient
                    .builder()
                    .serviceUrl("pulsar://123.114.60.34:16339")
                    .authentication(AuthenticationFactory.token(
                            "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxNjg0NzMzNjA4In0.Kh4Dlb5LibWJIa-hrFF5Y1Rl-fiZapqhp8YAoEPV2nA"))
                    .keepAliveInterval(5, TimeUnit.SECONDS)
                    .operationTimeout(10, TimeUnit.SECONDS)
                    .build();
        } catch (PulsarClientException e) {
            throw new RuntimeException(e);
        }
    }


    /*
     * @param void
     * @return 环控实体
     * */
    public static List<EnvironmentPojo> getEnvironment() {
        List<EnvironmentPojo> env = null;
        try {
            Consumer<byte[]> consumer = pulsarClient
                    .newConsumer()
                    .topic(AdmqClient.TopicEnum.topicEnv.producer.getTopic())
                    .subscriptionName("abc").subscribe();
            // 获取消息
            Message<byte[]> message = consumer.receive();
            // 获取页面对象,打印信息
            String msg = new String(consumer.receive().getValue());
            // 消息队列确认收到
            consumer.acknowledge(message);
            env = mapper.readValue(msg, new TypeReference<List<EnvironmentPojo>>() {});

            consumer.close();
        } catch (JsonProcessingException | PulsarClientException e) {
            throw new RuntimeException(e);
        }
        return env;
    }

    /*
     * @param void
     * @return 用水量实体
     * */
    public static List<WaterPojo> getWater() {
        List<WaterPojo> water = null;
        try {
            Consumer<byte[]> consumer = pulsarClient
                    .newConsumer()
                    .topic(AdmqClient.TopicEnum.topicWater.producer.getTopic())
                    .subscriptionName("abc").subscribe();
            // 获取消息
            Message<byte[]> message = consumer.receive();
            // 获取页面对象,打印信息
            String msg = new String(consumer.receive().getValue());
            // 消息队列确认收到
            consumer.acknowledge(message);
            water = mapper.readValue(msg, new TypeReference<List<WaterPojo>>() {});

            consumer.close();
        } catch (JsonProcessingException | PulsarClientException e) {
            throw new RuntimeException(e);
        }
        return water;
    }

    /*
     * @param void
     * @return 用电量实体
     * */
    public static List<ElectricityPojo> getElectricity() {
        List<ElectricityPojo> ele = null;
        try {
            Consumer<byte[]> consumer = pulsarClient
                    .newConsumer()
                    .topic(AdmqClient.TopicEnum.topicEle.producer.getTopic())
                    .subscriptionName("abc").subscribe();
            // 获取消息
            Message<byte[]> message = consumer.receive();
            // 获取页面对象,打印信息
            String msg = new String(consumer.receive().getValue());
            // 消息队列确认收到
            consumer.acknowledge(message);
            ele = mapper.readValue(msg, new TypeReference<List<ElectricityPojo>>() {});

            consumer.close();
        } catch (JsonProcessingException | PulsarClientException e) {
            throw new RuntimeException(e);
        }
        return ele;
    }




}
